package com.jfm.jfmc.core.icc.worker;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jfm.jfmc.core.icc.policy.WorkerDiscardPolicy;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * @author DYang
 * @since 2024-10-11 21:41
 */
@Slf4j
@Component
public class WorkerPool implements DisposableBean {

    @Value("${collector.worker.core-pool-size:10}")
    private int corePoolSize;

    @Value("${collector.worker.max-pool-size:100}")
    private int maxPoolSize;

    private ThreadPoolExecutor workerExecutor;


    @PostConstruct
    public void init(){
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setUncaughtExceptionHandler((thread, throwable) -> {
                    log.error("[Important] WorkerPool workerExecutor has uncaughtException.", throwable);
                    log.error("Thread Name {} : {}", thread.getName(), throwable.getMessage(), throwable);
                })
                .setDaemon(true)
                .setNameFormat("collect-worker-%d")
                .build();
        workerExecutor = new ThreadPoolExecutor(corePoolSize,
                maxPoolSize,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(100000),
                threadFactory,
                new WorkerDiscardPolicy());
    }

    /**
     * 执行任务
     * @param runnable
     * @throws RejectedExecutionException
     */
    public void executeJob(Runnable runnable) throws RejectedExecutionException {
        Runnable task = WorkerDiscardPolicy.getNextTask(runnable);
        if (task == null) {
            return;
        }
        workerExecutor.execute(task);
    }

    /**
     * 执行任务
     * @param runnable
     * @throws RejectedExecutionException
     */
    public Future<?> submitJob(Runnable runnable) throws RejectedExecutionException {
        Runnable task = WorkerDiscardPolicy.getNextTask(runnable);
        if (task == null) {
            return null;
        }
        return workerExecutor.submit(task);
    }

    @Override
    public void destroy() throws Exception {
        if (workerExecutor != null) {
            workerExecutor.shutdownNow();
        }
    }

    /**
     * 获取当前活跃线程数
     * @return
     */
    public int getActiveCount() {
        return workerExecutor.getActiveCount();
    }

    /**
     * 获取当前等待队列长度
     * @return
     */
    public int getQueueSize() {
        return workerExecutor.getQueue().size();
    }

    /**
     * 获取已完成的任务数量
     */
    public long getCompletedTaskCount() {
        return workerExecutor.getCompletedTaskCount();
    }
}
